package DianShang_2024.ds_03.clean

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{SparkSession, functions}
import org.apache.spark.sql.functions.{col, current_timestamp, date_format, desc, lit, min, row_number, to_timestamp}

import java.text.SimpleDateFormat
import java.util.Date

object clean02 {
  def main(args: Array[String]): Unit = {
    /*
          2、抽取ods库sku_info表中昨天的分区（子任务一生成的分区）数据，并结合dim_sku_info最新分区现有的数据，根据id合并数据到dwd库中
          dim_sku_info的分区表（合并是指对dwd层数据进行插入或修改，需修改的数据以id为合并字段，根据create_time排序取最新的一条），分区字段
          为etl_date且值与ods库的相对应表该值相等，并添加dwd_insert_user、dwd_insert_time、dwd_modify_user、dwd_modify_time四列,
          其中dwd_insert_user、dwd_modify_user均填写“user1”。若该条数据第一次进入数仓dwd层则dwd_insert_time、dwd_modify_time均填
          写当前操作时间，并进行数据类型转换。若该数据在进入dwd层时发生了合并修改，则dwd_insert_time时间不变，dwd_modify_time存当前操作时
          间，其余列存最新的值。使用hive cli查询表dim_sku_info的字段id、sku_desc、dwd_insert_user、dwd_modify_time、etl_date，条件
          为最新分区的数据，id大于等于15且小于等于20，并且按照id升序排序，将结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下；
     */
    //  准备环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("数据清洗第二题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use dwd03")

    //  1.使用sql的方式

//    //   读取到ods层的数据，并且对数据进行处理,下面的分区使用的是旧分区的数据
//    spark.sql(
//      """
//        |select
//        |id,
//        |spu_id,
//        |price,
//        |sku_name,
//        |sku_desc,
//        |weight,
//        |tm_id,
//        |category3_id,
//        |sku_default_img,
//        |create_time,
//        |etl_date,
//        |"user1" as dwd_insert_user,
//        |cast(to_timestamp(date_format(current_timestamp(),'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd HH:mm:ss') as timestamp) as dwd_insert_time,
//        |"user1" as dwd_modify_user,
//        |cast(to_timestamp(date_format(current_timestamp(),'yyyy-MM-dd HH:mm:ss'),'yyyy-MM-dd HH:mm:ss') as timestamp) as dwd_modify_time
//        |from ods03.sku_info
//        |where etl_date='20240101'
//        |""".stripMargin).createOrReplaceTempView("ods")
//
//
//    //  读取dwd层的数据，下面的分区的值为最新分区的值
//    spark.sql(
//      """
//        |select
//        |*
//        |from dwd03.dim_sku_info
//        |where etl_date='20240101'
//        |""".stripMargin).createOrReplaceTempView("dwd")
//
//    //  对两张表的数据进行合并,并且开窗函数进行数据合并
//    val result=spark.sql(
//      """
//        |select
//        |*
//        |from(
//        |select
//        |id,
//        |spu_id,
//        |price,
//        |sku_name,
//        |sku_desc,
//        |weight,
//        |tm_id,
//        |category3_id,
//        |sku_default_img,
//        |create_time,
//        |dwd_insert_user,
//        |cast(min(dwd_insert_time) over(partition by id) as timestamp) as dwd_insert_time,
//        |dwd_modify_user,
//        |cast(dwd_modify_time as timestamp),
//        |row_number() over(partition by id order by create_time desc ) as row,
//        |etl_date
//        |from(
//        |select * from dwd
//        |union all
//        |select * from ods
//        |) as t1
//        |) as t2
//        |where row=1
//        |""".stripMargin)
//
//    //   将row字段删掉
//    result.drop("row").drop("etl_date").createOrReplaceTempView("result")
//
//    //  插入数据
//    spark.sql(
//      """
//        |insert into table dwd03.dim_sku_info
//        |partition(etl_date='20240101')
//        |select * from result
//        |""".stripMargin)


    //  2.使用dataframe的方式
    //  读取到ods层的数据，并且对数据进行处理,这里的分区为旧分区的值
    val ods=spark.sql("select * from ods03.sku_info where etl_date='20240101'")
      .drop("etl_date")
      .withColumn("dwd_insert_user",lit("user1"))
      .withColumn(
        "dwd_insert_time",
        lit(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())).cast("timestamp")
      )
      .withColumn("dwd_modify_user",lit("user1"))
      .withColumn(
        "dwd_modify_time",
        lit(to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss"))
      )

    //   读取到dwd层的数据,分区的值为最新的分区值
    val dwd=spark.sql("select * from dwd03.dim_sku_info where etl_date='20240101'")
      .drop("etl_date")

    //  将两张表合并，使用开窗函数
    dwd.unionAll(ods)
      .withColumn(
        "dwd_insert_time",
        min("dwd_insert_time") over(Window.partitionBy("id"))
      )
      .withColumn(
        "row",
        row_number() over(Window.partitionBy("id").orderBy(desc("create_time")))
      )
      .where(col("row")===1)
      .drop("row")
      .withColumn("etl_date",lit("20240101"))
      .write.mode("append")
      .format("hive")
      .partitionBy("etl_date")
      .saveAsTable("dim_sku_info")



    //  关闭环境
    spark.close()
  }

}
